package cn.edu.fzu.delivery.session;

import java.util.Arrays;
import java.util.Collections;
import java.util.Enumeration;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;

import javax.servlet.ServletContext;
import javax.servlet.http.HttpSession;
import javax.servlet.http.HttpSessionContext;

import cn.edu.fzu.delivery.common.RedisClient;
import org.apache.commons.collections.MapUtils;

import com.alibaba.fastjson.JSON;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.pool.KryoFactory;
import com.esotericsoftware.kryo.pool.KryoPool;
import com.google.common.collect.Maps;
import org.apache.commons.lang3.ArrayUtils;

/**
 * <h2>getAttribute</h2>
 * <p>
 * 优先从本地缓存获取, 若不存在对应的 key, 从 redis 获取, 并刷新本地缓存
 * <h2>setAttribute</h2>
 * <p>
 * 对本地缓存进行设置, 如果值为 null, 需要记录到待删除列表
 * <hr>
 * <p>
 * 过滤器最后会将本地缓存持久化到 redis, 并将待删除的属性从 redis 删除
 * 
 * @author fangxin
 * @date 2022年11月3日
 */
public class RedisSession implements HttpSession {

    private static final int         SIZE      = 64;
    private static final int         MAX_SIZE  = 64 * 1024;

    private static final KryoPool    KRYO_POOL = new KryoPool.Builder(
            // factory
            new KryoFactory() {
                public Kryo create() {
                    return new Kryo();
                }
            }).build();

    // 缓存属性值, 减少 redis 访问次数
    private final Map<String, Value> cache     = Maps.newConcurrentMap();

    private final ServletContext     servletContext;
    private final RedisClient        redisClient;

    private final String             id;
    private final long               creationTime;
    private long                     lastAccessedTime;
    private int                      maxInactiveInterval;
    private boolean                  invalid;

    RedisSession(RedisClient redisClient, ServletContext servletContext, String id) {
        this.servletContext = servletContext;
        this.redisClient = redisClient;

        this.id = id;

        this.creationTime = System.currentTimeMillis();
        this.lastAccessedTime = this.creationTime;
        this.maxInactiveInterval = SessionConstant.SESSION_TIMEOUT;
        this.invalid = false;
    }


    @Override
    public Object getAttribute(String name) {

        if (invalid) {
            return null;
        }

        Value value = cache.get(name);
        if (value == null) {
            value = getAttributeFromRedis(name);
            cache.put(name, value);
        }

        return value.local;
    }


    @Override
    public void setAttribute(String name, Object value) {

        if (invalid) {
            return;
        }

        cache.put(name, Value.local(value));
    }


    @Override
    public void invalidate() {

        if (invalid) {
            return;
        }

        cache.clear();

        invalid = true;
    }


    void commit() {

        lastAccessedTime = System.currentTimeMillis();

        Key key = getSessionKey();

        if (MapUtils.isNotEmpty(cache)) {
            for (Entry<String, Value> entry : cache.entrySet()) {
                setOrDel(key, entry.getKey(), entry.getValue());
            }
            cache.clear();
        }

        if (invalid) {
            redisClient.delete(key.string);
        } else {
            redisClient.expire(key.string, getMaxInactiveInterval(), TimeUnit.SECONDS);
        }
    }


    /**
     * 从 redis 里获取属性值
     * 
     * @param name
     * @return Value.DUAL_NULL 表示没有这个属性
     */
    private Value getAttributeFromRedis(String name) {
        Key key = getSessionKey();

        byte[] value = (byte[]) redisClient.opsForHash().get(key.string, name);
        if (ArrayUtils.isEmpty(value)) {
            return Value.DUAL_NULL;
        }

        Kryo kryo = KRYO_POOL.borrow();
        try (Input input = new Input(value)) {
            return new Value(kryo.readClassAndObject(input), value);
        } finally {
            KRYO_POOL.release(kryo);
        }
    }


    // redis 赋值或删除
    private void setOrDel(Key key, String name, Value value) {

        if (value.local == null) {
            if (value.redis != null) {
                redisClient.opsForHash().delete(key.string, name);
            }
            return;
        }

        Kryo kryo = KRYO_POOL.borrow();
        try (Output output = new Output(SIZE, MAX_SIZE)) {
            kryo.writeClassAndObject(output, value.local);
            if (!Arrays.equals(value.redis, output.getBuffer())) {
                redisClient.opsForHash().put(key.string, name, JSON.toJSONString(output.getBuffer()));
            }
        } finally {
            KRYO_POOL.release(kryo);
        }
    }


    private final Key getSessionKey() {
        return new Key("sid:" + getId());
    }

    private static final class Key {

        private final String string;
        private final byte[] bytes;

        private Key(String key) {
            this.string = key;
            this.bytes = key.getBytes();
        }


        @Override
        public String toString() {
            return string;
        }
    }

    /**
     * 通过比较 local 和 redis 的值, 来确定是否需要对 redis 操作, 及何种操作
     * <ul>
     * <li>local==null && redis==null: 不操作</li>
     * <li>local==null && redis!=null: hdel</li>
     * <li>local!=null && redis!=null: local != redis --> hset</li>
     * <li>local!=null && redis==null: 非法取值</li>
     * </ul>
     * 注意： 除非明确查过 redis 才能将该 value.redis 设置为 null, 默认应该设置为 {@link Value#UNKNOWN}
     */
    private static final class Value {

        private static final byte[] UNKNOWN   = new byte[0];

        private final Object        local;
        private final byte[]        redis;

        private static final Value  DUAL_NULL = new Value(null, null);

        private Value(Object local, byte[] redis) {
            if (local != null && redis == null) {
                throw new IllegalArgumentException();
            }
            this.local = local;
            this.redis = redis;
        }


        private static final Value local(Object local) {
            return new Value(local, UNKNOWN);
        }


        @Override
        public String toString() {
            return JSON.toJSONString(local);
        }
    }

    @Override
    public String getId() {
        return id;
    }


    @Override
    public long getCreationTime() {
        return creationTime;
    }


    @Override
    public long getLastAccessedTime() {
        return lastAccessedTime;
    }


    @Override
    public ServletContext getServletContext() {
        return servletContext;
    }


    @Override
    public Enumeration<String> getAttributeNames() {
        return Collections.enumeration(cache.keySet());
    }


    @Override
    public void setMaxInactiveInterval(int interval) {
        this.maxInactiveInterval = interval;
    }


    @Override
    public int getMaxInactiveInterval() {
        return maxInactiveInterval;
    }


    @Override
    public boolean isNew() {
        return creationTime == lastAccessedTime;
    }


    @Override
    public void removeAttribute(String name) {
        setAttribute(name, null);
    }


    @Override
    public String[] getValueNames() {
        throw new UnsupportedOperationException();
    }


    @Override
    public Object getValue(String name) {
        throw new UnsupportedOperationException();
    }


    @Override
    public void putValue(String name, Object value) {
        throw new UnsupportedOperationException();
    }


    @Override
    public void removeValue(String name) {
        throw new UnsupportedOperationException();
    }


    @Override
    public HttpSessionContext getSessionContext() {
        throw new UnsupportedOperationException();
    }
}
